package com.rem.flink.flink10Sql;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 将 处理dataStream的环境 转换 处理table 的环境
 * Table 对象的基本查询操作 和基于环境的查询操作
 * table转换成 datastream 打印
 *
 * @author Rem
 * @date 2022-10-27
 */

public class SimpleTableExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1 获取数据源
        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        //2 获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //3 数据流转换成表
        Table eventTable = tableEnv.fromDataStream(stream);

        // 4.1 用执行SQL 的方式提取数据
        Table resultTable1 = tableEnv.sqlQuery("select url, user, `timestamp` from " + eventTable);

        // 4.2 基于Table直接转换
        Table resultTable2 = eventTable.select($("user"), $("url"), $("timestamp")).where($("user").isEqual("Alice"));

        tableEnv.toDataStream(resultTable1).print("result1:");
        tableEnv.toDataStream(resultTable2).print("result2:");
        env.execute();
    }
}
